package com.typhoon.spring_shiro.service.impl;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericToStringSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.fastjson.JSON;
import com.typhoon.spring_shiro.constants.CenterEntityError;
import com.typhoon.spring_shiro.dao.StockDao;
import com.typhoon.spring_shiro.dao.StockOptHistoryDao;
import com.typhoon.spring_shiro.dto.StockDto;
import com.typhoon.spring_shiro.dto.request.StockReq;
import com.typhoon.spring_shiro.dto.response.StockResp;
import com.typhoon.spring_shiro.entity.Stock;
import com.typhoon.spring_shiro.entity.StockOptHistory;
import com.typhoon.spring_shiro.exception.ServiceRuntimeException;
import com.typhoon.spring_shiro.service.StockService;
import com.typhoon.spring_shiro.utils.BeanUtils;
import com.typhoon.spring_shiro.utils.DateUtil;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

/**
 * StockService接口默认实现
 * 
 * @author Typhoon
 * @date 2017-07-11 12:05
 * @since V1.3.0
 */
@Service("stockService")
public class StockServiceImpl implements StockService {

	private static final Logger LOGGER = LoggerFactory.getLogger(StockServiceImpl.class);

	@Resource
	StockDao stockDao;

	@Resource
	StockOptHistoryDao stockOptHistoryDao;

	@Resource
	ThreadPoolTaskExecutor taskExecutor;

	@Resource
	RedisTemplate<String, Integer> redisTemplate;

	private RedisSerializer<Integer> valueSerializer = new GenericToStringSerializer<Integer>(Integer.class);

	@Resource
	JedisPool jedisPool;
	
	@Override
	@Transactional
	public int save(StockDto paramE) {
		if (null == paramE) {
			return -1;
		}
		Stock stock = this.stockDao.save(BeanUtils.toBean(paramE, Stock.class));
		if (null == stock || null == stock.getId()) {
			throw new ServiceRuntimeException("保存失败");
		}
		return 1;
	}

	@Override
	public int saveBatch(List<? extends StockDto> paramList) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public int update(StockDto paramE) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public int updateBatch(List<StockDto> paramList) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public int deleteByPK(Long paramK) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public int deleteBatchByPK(List<Long> paramList) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public int delete(StockDto paramE) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public int deleteBatch(StockDto paramE) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public StockDto get(Object paramObject) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public List<StockDto> list(Map<String, Object> paramMap) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public List<StockDto> list(Object paramObject) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public long count(Map<String, Object> paramMap) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	public long count(Object paramObject) {
		throw new UnsupportedOperationException("this method is not realized");
	}

	@Override
	@Transactional
	public StockResp deductStock(StockReq req) {
		LOGGER.info("扣减库存请求参数:{}", JSON.toJSONString(req));
		StockResp resp = new StockResp();
		resp.setThreadName(Thread.currentThread().getName());
		Calendar cal = Calendar.getInstance();
		if (null == req || null == req.getId() || null == req.getNeedDeduct()) {
			// 参数非法
			return resp.error(CenterEntityError.ILLEGAL_ARGUMENT_ERROR);
		}
		Stock stock = this.stockDao.queryByPrimaryKey(req.getId());
		LOGGER.info("id:{}对应库存记录信息:{}", req.getId(), JSON.toJSONString(stock));
		if (null == stock) {// id对应stock不存在
			LOGGER.info("id:{}对应stock不存在", req.getId());
			return resp.error("stock不存在");
		}
		if (stock.getStock() <= 0 || stock.getStock() < req.getNeedDeduct()) {// 不足以扣减库存
			LOGGER.info("id:{}库存为:{},需要扣减的库存:{},不足以抵扣", req.getId(), stock.getStock(), req.getNeedDeduct());
			return resp.error("库存不足");
		}
		Map<String, Object> params = new HashMap<>(5);
		params.put("id", req.getId());
		params.put("oldStock", stock.getStock());
		params.put("newStock", stock.getStock() - req.getNeedDeduct());// 把扣减库存转换成设置库存,由非幂等操作转换成幂等操作
		params.put("version", stock.getVersion());
		params.put("updateTime", DateUtil.formatDate(cal.getTime(), "yyyy-MM-dd HH:mm:ss"));
		int rs = this.stockDao.adjustStock(params);
		LOGGER.info("id:{}扣减库存dao操作结果:{}", req.getId(), rs);
		if (rs <= 0) {// 如果没有更新成功,按常理说需要抛RuntimeException回滚,但是只有一个更新操作直接返回错误码即可
			// throw new ServiceRuntimeException("id:"+ req.getId() + "扣减库存失败");
			LOGGER.info("id:{}扣减库存失败", req.getId());
			return resp.error("扣减库存失败");
		}
		return resp;
	}

	@Override
	@Transactional
	public StockResp increaseStock(StockReq req) {
		LOGGER.info("增加库存请求参数:{}", JSON.toJSONString(req));
		StockResp resp = new StockResp();
		resp.setThreadName(Thread.currentThread().getName());
		Calendar cal = Calendar.getInstance();
		if (null == req || null == req.getId() || null == req.getNeedIncrease()) {
			// 参数非法
			return resp.error(CenterEntityError.ILLEGAL_ARGUMENT_ERROR);
		}
		Stock stock = this.stockDao.queryByPrimaryKey(req.getId());
		LOGGER.info("id:{}对应库存记录信息:{}", req.getId(), JSON.toJSONString(stock));
		if (null == stock) {// id对应stock不存在
			LOGGER.info("id:{}对应stock不存在", req.getId());
			return resp.error("stock不存在");
		}
		Map<String, Object> params = new HashMap<>(5);
		params.put("id", req.getId());
		params.put("oldStock", stock.getStock());
		params.put("newStock", stock.getStock() + req.getNeedIncrease());// 把增加库存转换成设置库存,由非幂等操作转换成幂等操作
		params.put("version", stock.getVersion());
		params.put("updateTime", DateUtil.formatDate(cal.getTime(), "yyyy-MM-dd HH:mm:ss"));
		int rs = this.stockDao.adjustStock(params);
		LOGGER.info("id:{}增加库存dao操作结果:{}", req.getId(), rs);
		if (rs <= 0) {// 如果没有更新成功,按常理说需要抛RuntimeException回滚,但是只有一个更新操作直接返回错误码即可
			// throw new ServiceRuntimeException("id:"+ req.getId() + "增加库存失败");
			LOGGER.info("id:{}增加库存失败", req.getId());
			return resp.error("增加库存失败");
		}
		return resp;
	}

	@Override
	@Transactional
	public StockResp deductStockV2(StockReq req) {
		LOGGER.info("当前线程:{},V2扣减库存请求参数:{}", Thread.currentThread().getName(), JSON.toJSONString(req));
		StockResp resp = new StockResp();
		resp.setThreadName(Thread.currentThread().getName());
		Calendar cal = Calendar.getInstance();
		if (null == req || null == req.getId() || null == req.getNeedDeduct()) {
			// 参数非法
			return resp.error(CenterEntityError.ILLEGAL_ARGUMENT_ERROR);
		}
		Stock stock = this.stockDao.queryByPrimaryKey(req.getId());
		LOGGER.info("id:{}对应库存记录信息:{},当前线程:{}", req.getId(), JSON.toJSONString(stock), Thread.currentThread().getName());
		if (null == stock) {// id对应stock不存在
			LOGGER.info("id:{}对应stock不存在", req.getId());
			return resp.error("stock不存在");
		}

		// ①先查询库存并做比较;实际库存 > redis中待扣减库存 + 本次扣减库存,才会执行后续扣减动作
		int currentWaitingForDeduct = this.getStockWaitingForDeduct(req.getId());
		int stockSku = stock.getStock();
		if (stockSku < currentWaitingForDeduct + req.getNeedDeduct()) {
			LOGGER.info("当前线程:{},id:{}当前库存:{},缓存中待扣减:{},需要扣减:{},库存不足", Thread.currentThread().getName(), req.getId(), stockSku, currentWaitingForDeduct,
					req.getNeedDeduct());
			return resp.error("库存不足");
		}

		// 如果有足额库存,本次扣减数量 在扣减之前锁定到待派发数量缓存中
		long skuLock = this.handleStockWaitingForDeduct(req.getId(), req.getNeedDeduct());
		LOGGER.info("当前线程:{},id:{}加库存锁定后待扣减数量:{}", Thread.currentThread().getName(), req.getId(), skuLock);

		stock.setStock(stock.getStock() - req.getNeedDeduct());
		stock.setUpdateTime(cal.getTime());
		int rs = this.stockDao.update(stock);
		LOGGER.info("当前线程:{},id:{}扣减库存dao操作结果:{}", Thread.currentThread().getName(), req.getId(), rs);
		if (rs <= 0) {// 如果没有更新成功,按常理说需要抛RuntimeException回滚,但是只有一个更新操作直接返回错误码即可
			// throw new ServiceRuntimeException("id:"+ req.getId() + "扣减库存失败");
			LOGGER.info("当前线程:{},id:{}扣减库存失败", Thread.currentThread().getName(), req.getId());
			return resp.error("扣减库存失败");
		}
		this.handleStockWaitingForDeduct(req.getId(), -req.getNeedDeduct());
		return resp;
	}

	@Override
	public int getStockWaitingForDeduct(Long stockId) {
		StringBuilder keyBuilder = new StringBuilder(CACHE_STOCK_WAITING_FOR_DEDUCT_PREFIX);
		keyBuilder.append(stockId);
		LOGGER.info("库存待扣减前缀:{}", keyBuilder.toString());
		Integer value = 0;

		redisTemplate.setValueSerializer(valueSerializer);
		try {
			value = redisTemplate.opsForValue().get(keyBuilder.toString());
			// value = RedisUtils.getIntOfStr(keyBuilder.toString());
		} catch (Exception e) {
			LOGGER.error("获取库存待扣减数量异常", e);
			// RedisUtils.setIntOfStr(keyBuilder.toString(), value,
			// ONE_DAY_SECONDS);
			redisTemplate.opsForValue().set(keyBuilder.toString(), value, 24, TimeUnit.HOURS);
		}
		if (null == value) {
			value = 0;
		}
		LOGGER.info("id:{}的当前待扣减库存:{}", stockId, value);
		return value;
	}

	@Override
	public long handleStockWaitingForDeduct(Long stockId, int amount) {
		StringBuilder keyBuilder = new StringBuilder(CACHE_STOCK_WAITING_FOR_DEDUCT_PREFIX);
		keyBuilder.append(stockId);
		LOGGER.info("库存待扣减前缀:{}", keyBuilder.toString());
		long result = 0;

		redisTemplate.setValueSerializer(valueSerializer);
		// if (null == redisTemplate.opsForValue().get(keyBuilder.toString())) {
		// return result;
		// }
		result = redisTemplate.opsForValue().increment(keyBuilder.toString(), amount);
		if (result < 0) {
			redisTemplate.opsForValue().set(keyBuilder.toString(), 0, 24, TimeUnit.HOURS);
		}
		result = redisTemplate.opsForValue().get(keyBuilder.toString());
		LOGGER.info("id:{}当前库存待扣减数量为:{}", stockId, result);
		return result;
	}

	@Override
	public StockResp deductStockV3(StockReq req) {
		LOGGER.info("当前线程:{},V2扣减库存请求参数:{}", Thread.currentThread().getName(), JSON.toJSONString(req));
		StockResp resp = new StockResp();
		resp.setThreadName(Thread.currentThread().getName());
		Calendar cal = Calendar.getInstance();
		if (null == req || null == req.getId() || null == req.getNeedDeduct() || null == req.getCustomerId()) {
			// 参数非法
			return resp.error(CenterEntityError.ILLEGAL_ARGUMENT_ERROR);
		}
		
		//限制同一个客户端5秒请求一次
		Jedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			String rateLimitScript = buildRateLimitScript();
			List<String> keys = new ArrayList<>(1);
			keys.add(CACHE_STOCK_WAITING_FOR_DEDUCT_PREFIX + req.getId() + ":" + req.getCustomerId());
			long result = Long.parseLong(jedis.eval(rateLimitScript, keys, keys) + "");
			if(result <= 0) {//超过访问限制
				return resp.error("超过访问限制次数");
			}
		} catch (Exception e) {
			LOGGER.error("",e);
			return resp.error("异常");
		} finally {
			IOUtils.closeQuietly(jedis);
		}
		
		Stock stock = this.stockDao.queryByPrimaryKey(req.getId());// 两个线程都查到库存为4
		LOGGER.info("id:{}对应库存记录信息:{},当前线程:{}", req.getId(), JSON.toJSONString(stock), Thread.currentThread().getName());
		if (null == stock) {// id对应stock不存在
			LOGGER.info("id:{}对应stock不存在", req.getId());
			return resp.error("stock不存在");
		}

		List<String> keys = new ArrayList<>();
		keys.add(CACHE_STOCK_WAITING_FOR_DEDUCT_PREFIX + req.getId());
		List<String> args = new ArrayList<>();
		args.add(req.getNeedDeduct() + "");
		args.add(stock.getStock() + "");

		try {
			jedis = jedisPool.getResource();
			// 使用redis + lua脚本控制查询的锁定缓存操作,能够保证原子操作
			// ①库存查询预锁定
			String luaLockScript = buildLuaLockScript();
			long result = Long.parseLong(jedis.eval(luaLockScript, keys, args) + "");
			// count = Long.parseLong(jedis.eval(buildLuaScript(limitRule), keys,
			// args) + "");
			LOGGER.info("线程:{}对id:{}锁定库存结果:{}", Thread.currentThread().getName(), req.getId(), result);
			if (result <= 0) {// 锁定失败
				return resp.error("库存不足");
			}
		} catch (Exception e) {
			LOGGER.error("",e);
			return resp.error("异常");
		} finally {
			if (jedis != null)
				jedis.close();
		}

		Map<String, Object> params = new HashMap<>(3);
		params.put("id", req.getId());
		params.put("needDeduct", req.getNeedDeduct());
		params.put("updateTime", DateUtil.formatDate(cal.getTime(), "yyyy-MM-dd HH:mm:ss"));

		int rs = this.stockDao.deductStock(params);
		LOGGER.info("当前线程:{},id:{}扣减库存dao操作结果:{}", Thread.currentThread().getName(), req.getId(), rs);
		// ②解除库存锁定,不管实际库存扣减成功和失败,都要解除库存锁定,因为既然能够走到db层扣减库存,那么前一步的库存锁定肯定已经成功了
		args = new ArrayList<>(2);
		args.add(req.getNeedDeduct() + "");
		args.add(3600 + "");
		try {
			jedis = jedisPool.getResource();
			String luaUnlockScript = buildLuaUnlockScript();
			long result = Long.parseLong(jedis.eval(luaUnlockScript, keys, args) + "");
			LOGGER.info("线程:{}对id:{}库存解除锁定结果:{}", Thread.currentThread().getName(), req.getId(), result);
		} catch (Exception e) {
			LOGGER.error("",e);
			return resp.error("异常");
		}finally {
			if (jedis != null)
				jedis.close();
		}
		if (rs <= 0) {// 如果没有更新成功,按常理说需要抛RuntimeException回滚,但是只有一个更新操作直接返回错误码即可
			// throw new ServiceRuntimeException("id:"+ req.getId() + "扣减库存失败");
			LOGGER.info("当前线程:{},id:{}扣减库存失败", Thread.currentThread().getName(), req.getId());
			return resp.error("扣减库存失败");
		}

		StockOptHistory history = new StockOptHistory();
		history.setStockId(req.getId());
		history.setCreateTime(cal.getTime());
		history.setDeductStock(req.getNeedDeduct());
		history.setOldStock(stock.getStock());
		history.setThreadName(Thread.currentThread().getName());
		history.setNewStock(stock.getStock() - req.getNeedDeduct());
		taskExecutor.execute(new HistoryRunnable(history, stockOptHistoryDao, LOGGER));

		return resp;
	}
	//
	// private static String buildLuaScript() {
	// StringBuilder lua = new StringBuilder();
	// lua.append("\nlocal c");
	// lua.append("\nc = redis.call('decrby',KEYS[1],ARGV[1])");//解除锁定
	// lua.append("\nreturn c;");
	// return lua.toString();
	// }

	/**
	 * 访问速率限制lua脚本
	 * 
	 * @author Typhoon
	 * @date 2017-07-13 11:07 Thursday
	 * @since V1.3.0
	 * @return
	 */
	private String buildRateLimitScript() {
		StringBuilder lua = new StringBuilder();
		lua.append("\nlocal c");
		lua.append("\nc = redis.call('setnx',KEYS[1],1)");
		lua.append("\nif c and tonumber(c) > 0 then");//锁定成功
		lua.append("\nredis.call('expire',KEYS[1],5)");//设置5秒过期
		lua.append("\nend");
		lua.append("\nreturn c;");
		return lua.toString();
	}

	/**
	 * 解除库存锁定lua脚本
	 * 
	 * @author Typhoon
	 * @date 2017-07-12 14:39 Wednesday
	 * @since V1.3.0
	 * @return
	 */
	private String buildLuaUnlockScript() {
		StringBuilder lua = new StringBuilder();
		lua.append("\nlocal c");
		lua.append("\nc = redis.call('decrby',KEYS[1],ARGV[1])");// 解除锁定
		lua.append("\nif c and tonumber(c) < 0 then");
		lua.append("\nredis.call('expire',KEYS[1],ARGV[2])");// 设置过期时间
		lua.append("\nend");
		lua.append("\nreturn c;");
		return lua.toString();
	}

	/**
	 * 查询&锁定库存lua脚本
	 * 
	 * @author Typhoon
	 * @date 2017-07-12 14:39 Wednesday
	 * @since V1.3.0
	 * @return
	 */
	private static String buildLuaLockScript() {
		StringBuilder lua = new StringBuilder();
		lua.append("\nlocal c");
		lua.append("\nc = redis.call('get',KEYS[1])");
		lua.append("\nif c then");// 如果对应key不存在,c默认为0
		lua.append("\nc = tonumber(c)");
		lua.append("\nelse");
		lua.append("\nc = 0");
		lua.append("\nend");// c完成初始化

		lua.append("\nif (tonumber(c) + tonumber(ARGV[1])) > tonumber(ARGV[2]) then");// cache
																						// +
																						// needDeduct
																						// >
																						// stock
		lua.append("\nc = -1");// 返回负值
		lua.append("\nelse ");// 第一次操作库存,key不存在
		lua.append("\nc = redis.call('incrby',KEYS[1],ARGV[1]) ");// 锁定库存
		lua.append("\nend");
		lua.append("\nreturn c;");
		return lua.toString();
	}

	class HistoryRunnable implements Runnable {

		private StockOptHistory history;

		private StockOptHistoryDao stockOptHistoryDao;

		private Logger log;

		public HistoryRunnable(StockOptHistory history, StockOptHistoryDao stockOptHistoryDao, Logger log) {
			super();
			this.history = history;
			this.stockOptHistoryDao = stockOptHistoryDao;
			this.log = log;
		}

		public StockOptHistory getHistory() {
			return history;
		}

		public void setHistory(StockOptHistory history) {
			this.history = history;
		}

		public StockOptHistoryDao getStockOptHistoryDao() {
			return stockOptHistoryDao;
		}

		public void setStockOptHistoryDao(StockOptHistoryDao stockOptHistoryDao) {
			this.stockOptHistoryDao = stockOptHistoryDao;
		}

		public Logger getLog() {
			return log;
		}

		public void setLog(Logger log) {
			this.log = log;
		}

		@Override
		public void run() {
			history = this.stockOptHistoryDao.save(history);
			log.info("保存操作历史响应结果:{}", JSON.toJSONString(history));
		}

	}

	@Override
	public StockResp deductStockV4(StockReq req) {
		StockResp resp = new StockResp();
		//this.stockDao.queryByPrimaryKey(req.get)
//		Stock stock = redisDao.getStock(req.getId());
//		if(null == stock) {//缓存中没有
//			
//			stock = this.stockDao.queryByPrimaryKey(req.getId());
//			
//			if(null == stock) {
//				return resp.error("库存不存在");
//			} else {//缓存没有db有,把数据放入缓存
//				this.redisDao.putStock(stock);
//			}
//			
//		}
//		
//		//A:4,B:4
//		
//		if(stock.getStock() < req.getNeedDeduct()) {//库存不足额
//			return resp.error("库存不足额");
//		}
//		
//		stock.setStock(stock.getStock() - req.getNeedDeduct());
//		
//		//A:1,B:1
//		this.redisDao.putStock(stock);//覆盖,导致少扣
		
		return resp;
	}

	@Override
	public StockResp deductStockV5(StockReq req) {
		StockResp resp = new StockResp();
		long stockId = req.getId();
		Jedis jedis = null;
		long result = 0;
		Calendar cal = Calendar.getInstance();
		try {
			Stock stock = this.stockDao.queryByPrimaryKey(req.getId());
			if(null == stock) {
				return resp.error("不存在");
			}
			jedis = jedisPool.getResource();
			List<String> keys = new ArrayList<>(1);
			keys.add(CACHE_STOCK_WAITING_FOR_DEDUCT_PREFIX + stockId);
			List<String> args = new ArrayList<>(1);
			args.add(req.getNeedDeduct() + "");
			args.add(stock.getStock() + "");
			String deductLuaScript = buildLuaDeductScript();
			result = Long.parseLong(jedis.eval(deductLuaScript, keys, args) + "");
			if(result < 0) {//库存不足
				return resp.error("库存不足");
			}
			stock.setStock((int)result);
			stock.setUpdateTime(cal.getTime());
			this.stockDao.update(stock);
			
			//记录操作历史
			StockOptHistory history = new StockOptHistory();
			history.setStockId(req.getId());
			history.setCreateTime(cal.getTime());
			history.setDeductStock(req.getNeedDeduct());
			history.setOldStock(stock.getStock());
			history.setThreadName(Thread.currentThread().getName());
			history.setNewStock(stock.getStock() - req.getNeedDeduct());
			taskExecutor.execute(new HistoryRunnable(history, stockOptHistoryDao, LOGGER));
		} catch (Exception e) {
			//获取连接失败,或者其他异常,抛出去,用来终止程序
			throw new ServiceRuntimeException("异常",e);
		} finally {
			IOUtils.closeQuietly(jedis);
		}
		return resp;
	}
	
	/**
	 * 扣减库存lua脚本
	 * 
	 * @author Typhoon
	 * @date 2017-07-12 21:30
	 * @since V1.3.0
	 * @return
	 */
	private String buildLuaDeductScript() {
		StringBuilder lua = new StringBuilder();
		lua.append("\nlocal c");
		lua.append("\nredis.call('setnx',KEYS[1],ARGV[2])");//如果库存没有放到redis中放进去
		lua.append("\nc = redis.call('get',KEYS[1])");
		lua.append("\nif c and tonumber(c) < tonumber(ARGV[1]) then");//如果库存不足,stock < needDeduct
		lua.append("\nc = -1");
		lua.append("\nelse");//如果库存足额
		lua.append("\nc = redis.call('decrby',KEYS[1],ARGV[1])");//扣除库存
		lua.append("\nend");
		lua.append("\nreturn c;");
		return lua.toString();
	}

	@Override
	public StockDto queryByPK(Long paramK) {
		// TODO Auto-generated method stub
		return null;
	}

}
